package com.eatme.web.app.coupon.distribution.service.impl;

import com.alibaba.fastjson2.JSON;
import com.eatme.web.app.coupon.common.constant.Constant;
import com.eatme.web.app.coupon.common.exception.AppException;
import com.eatme.web.app.coupon.common.vo.CouponTemplateSDK;
import com.eatme.web.app.coupon.common.vo.GoodsInfo;
import com.eatme.web.app.coupon.common.vo.SettlementInfo;
import com.eatme.web.app.coupon.distribution.constant.CouponStatus;
import com.eatme.web.app.coupon.distribution.dao.CouponDao;
import com.eatme.web.app.coupon.distribution.entity.Coupon;
import com.eatme.web.app.coupon.distribution.feign.SettlementClient;
import com.eatme.web.app.coupon.distribution.service.IRedisService;
import com.eatme.web.app.coupon.distribution.service.IUserService;
import com.eatme.web.app.coupon.distribution.vo.AcquireTemplateRequest;
import com.eatme.web.app.coupon.distribution.vo.CouponClassify;
import com.eatme.web.app.coupon.distribution.vo.CouponKafkaMessage;
import com.eatme.web.app.coupon.distribution.feign.TemplateClient;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 用户服务相关接口实现
 * 所有的操作过程,状态都保存在Redis 中并通过Kafka 把消息传递到MySQL 中
 * 为什么使用Kafka, 而不是使用 SpringBoot 中的异步处理 ?
 */
@Service("couponUserServiceImpl")
public class UserServiceImpl implements IUserService {

    private static final Logger LOG = LoggerFactory.getLogger(UserServiceImpl.class);

    /** Coupon Dao */
    private final CouponDao couponDao;

    /** Redis 服务 */
    private final IRedisService redisService;


    private final KafkaTemplate<String,String> kafkaTemplate;

    private final TemplateClient templateClient;

    /** 结算微服务客户端 */
    private final SettlementClient settlementClient;

    public UserServiceImpl(CouponDao couponDao, IRedisService redisService, KafkaTemplate<String, String> kafkaTemplate, TemplateClient templateClient, SettlementClient settlementClient) {
        this.couponDao = couponDao;
        this.redisService = redisService;
        this.kafkaTemplate = kafkaTemplate;
        this.templateClient = templateClient;
        this.settlementClient = settlementClient;
    }


    @Override
    public List<Coupon> findCouponsByStatus(Long userId, Integer status) throws AppException {
        List<Coupon> curCached = redisService.getCacheCoupons(userId,status);
        List<Coupon> preTarget;
        if (CollectionUtils.isNotEmpty(curCached)) {
            LOG.debug("coupon cache is not empty: {}, {}",userId,status);
            preTarget = curCached;
        }else {
            LOG.info("coupon cache is empty, get coupon from db: {}, {}",userId,status);
            List<Coupon> dbCoupons = couponDao.findAllByUserIdAndStatus(userId, CouponStatus.of(status));
            // 如果数据库没有记录,直接返回就可以了,Cache 中已经加入了一张无效的优惠券
            if (CollectionUtils.isEmpty(dbCoupons)) {
                LOG.debug("current user not have coupon: {}, {}",userId,status);
                return dbCoupons;
            }

            // 填充dbCoupons的 templateSDK 字段
            Map<Integer, CouponTemplateSDK> id2TemplateSDK = templateClient.findIds2TemplateSDK(
                    dbCoupons.stream()
                            .map(Coupon::getTemplateId)
                            .collect(Collectors.toList())
            ).getData();

            dbCoupons.forEach(
                    dc -> dc.setTemplateSDK(id2TemplateSDK.get(dc.getTemplateId()))
            );

            // 数据库中存在记录
            preTarget =dbCoupons;

            // 将记录写入 Cache
            redisService.addCouponToCache(userId,preTarget,status);

        }

        // 将无效优惠券剔除
        preTarget = preTarget.stream().
                filter(coupon -> coupon.getId()!=-1)
                .collect(Collectors.toList());
        // 如果当前获取的是可用优惠券,还需要做对已过期优惠券的延迟处理
        if (CouponStatus.of(status)==CouponStatus.USABLE) {
            CouponClassify classify = CouponClassify.classify(preTarget);
            // 如果过期状态不为空,需要做延迟处理
            if (CollectionUtils.isNotEmpty(classify.getExpired())) {
                LOG.info("Add Expired Coupons To Cache From findCouponsByStatus: {}, {}"
                ,userId,status);
                redisService.addCouponToCache(
                        userId,
                        classify.getExpired(),
                        CouponStatus.EXPIRED.getCode()
                );
                // 发送到 kafka 中做异步处理
                kafkaTemplate.send(
                        Constant.TOPIC,
                        JSON.toJSONString(new CouponKafkaMessage(
                                CouponStatus.EXPIRED.getCode(),
                                classify.getExpired().stream()
                                        .map(Coupon::getId).collect(Collectors.toList())
                        ))
                );
            }
            return classify.getUsable();
        }
        return preTarget;
    }

    @Override
    public List<CouponTemplateSDK> findAvailableTemplate(Long userId) throws AppException {
        Long curTime = new Date().getTime();
        List<CouponTemplateSDK> templateSDKS =
                templateClient.findAllUsableTemplate().getData();

        LOG.debug("Find Template(From TemplateClient) Count: {}",
                templateSDKS.size());

        // 过滤过期的优惠券模版
        templateSDKS = templateSDKS.stream().filter(
                t -> t.getRule().getExpiration().getDeadLine() > curTime
        ).collect(Collectors.toList());

        LOG.info("Find Usable Template Count: {}",templateSDKS.size());

        // key 是 TemplateId
        // value 中的 left 是 Template limitation, value 是优惠券模版
        Map<Integer, Pair<Integer,CouponTemplateSDK>> limit2Template =
                new HashMap<>(templateSDKS.size());

        templateSDKS.forEach(t->limit2Template.put(
                t.getId(),
                Pair.of(t.getRule().getLimitation(),t)
                )
        );

        List<CouponTemplateSDK> result = new ArrayList<>(limit2Template.size());
        List<Coupon> userUsableCoupons = findCouponsByStatus(
                userId,CouponStatus.USABLE.getCode()
        );
        LOG.info("Current User Has Usable Coupons: {}, {}",userId,userUsableCoupons.size());

        // key 是 TemplateId
        Map<Integer,List<Coupon>> templateId2Coupons = userUsableCoupons.
                stream()
                .collect(Collectors.groupingBy(Coupon::getTemplateId));

        // 根据 Template 的 Rule 判断是否可以领取优惠券模版
        limit2Template.forEach((k,v) -> {
            int limitation = v.getLeft();
            CouponTemplateSDK templateSDK = v.getRight();
            if (templateId2Coupons.containsKey(k)
                    && templateId2Coupons.get(k).size() >= limitation) {
                return;
            }
            result.add(templateSDK);
        });

        return result;
    }

    /**
     * 用户领取优惠券
     * 1. 从TemplateClient 拿到对应的优惠券，并检查是否过期
     * 2. 根据 limitation 判断用户是否可以领取
     * 3. save to db
     * 4. 填充 CouponTemplateSDK
     * 5. save to cache
     * @param request {@link AcquireTemplateRequest}
     * @return {@link Coupon}
     * @throws AppException
     */
    @Override
    public Coupon acquireTemplate(AcquireTemplateRequest request)
            throws AppException {
        Map<Integer,CouponTemplateSDK> id2Template =
                templateClient.findIds2TemplateSDK(
                  Collections.singletonList(
                          request.getTemplateSDK().getId())
                ).getData();

        // 优惠券模版是需要存在的
        if (id2Template.size() <= 0) {
            LOG.error("Can Not Acquire Template From TemplateClient: {}",
                    request.getTemplateSDK().getId());
            throw new AppException("Can Not Acquire Template From TemplateClient");
        }

        // 用户是否可以领取这张优惠券
        List<Coupon> userUsableCoupons = findCouponsByStatus(
                request.getUserId(),CouponStatus.USABLE.getCode()
        );

        Map<Integer,List<Coupon>> templateId2Coupons = userUsableCoupons
                .stream()
                .collect(Collectors.groupingBy(Coupon::getTemplateId));
        if (templateId2Coupons.containsKey(request.getTemplateSDK().getId())
        &&templateId2Coupons.get(request.getTemplateSDK().getId()).size() >=
                request.getTemplateSDK().getRule().getLimitation()) {
            LOG.info("Exceed Template Assign limitation: {}",request.getTemplateSDK().getId());
            throw  new AppException("Exceed Template Assign limitation");

        }

        // 尝试去获取优惠券码
        String couponCode = redisService.tryToAcquireCouponCodeFromCache(
                request.getTemplateSDK().getId()
        );

        if (StringUtils.isEmpty(couponCode)) {
            LOG.error("Can Not Acquire Coupon Code");
        }

        Coupon newCoupon = new Coupon(
                request.getTemplateSDK().getId(),
                request.getUserId(),
                couponCode,CouponStatus.USABLE
        );
        newCoupon = couponDao.save(newCoupon);

        // 填充 Coupon 对象的 CouponTemplateSDK,一定要在放入缓存之前去填充
        newCoupon.setTemplateSDK(request.getTemplateSDK());

        //放入缓存中
        redisService.addCouponToCache(
                request.getUserId(),
                Collections.singletonList(newCoupon),
                CouponStatus.USABLE.getCode()
        );

        return newCoupon;
    }

    /**
     * <h2>结算(核销)优惠券</h2>
     * 这里需要注意, 规则相关处理需要由 Settlement 系统去做, 当前系统仅仅做
     * 业务处理过程(校验过程)
     * @param info {@link SettlementInfo}
     * @return {@link SettlementInfo}
     */
    @Override
    public SettlementInfo settlement(SettlementInfo info) throws AppException {

        // 当没有传递优惠券时, 直接返回商品总价
        List<SettlementInfo.CouponAndTemplateInfo> ctInfos =
                info.getCouponAndTemplateInfos();
        if (CollectionUtils.isEmpty(ctInfos)) {

            LOG.info("Empty Coupons For Settle.");

            // double goodsSum = 0.0;
            Integer goodsSum = 0;

            for (GoodsInfo gi : info.getGoodsInfos()) {
                goodsSum += gi.getPrice() + gi.getCount();
            }

            // 没有优惠券也就不存在优惠券的核销, SettlementInfo 其他的字段不需要修改

            // info.setCost(retain2Decimals(goodsSum));
            info.setCost(goodsSum);
        }

        // 校验传递的优惠券是否是用户自己的
        List<Coupon> coupons = findCouponsByStatus(
                info.getUserId(), CouponStatus.USABLE.getCode()
        );
        Map<Integer, Coupon> id2Coupon = coupons.stream()
                .collect(Collectors.toMap(
                        Coupon::getId,
                        Function.identity()
                ));
        if (MapUtils.isEmpty(id2Coupon) || !CollectionUtils.isSubCollection(
                ctInfos.stream().map(SettlementInfo.CouponAndTemplateInfo::getId)
                        .collect(Collectors.toList()), id2Coupon.keySet()
        )) {
            LOG.info("{}", id2Coupon.keySet());
            LOG.info("{}", ctInfos.stream()
                    .map(SettlementInfo.CouponAndTemplateInfo::getId)
                    .collect(Collectors.toList()));
            LOG.error("User Coupon Has Some Problem, It Is Not SubCollection" +
                    "Of Coupons!");
            throw new AppException("User Coupon Has Some Problem, " +
                    "It Is Not SubCollection Of Coupons!");
        }

        LOG.debug("Current Settlement Coupons Is User's: {}", ctInfos.size());

        List<Coupon> settleCoupons = new ArrayList<>(ctInfos.size());
        ctInfos.forEach(ci -> settleCoupons.add(id2Coupon.get(ci.getId())));

        // 通过结算服务获取结算信息
        SettlementInfo processedInfo =
                settlementClient.computeRule(info).getData();
        if (processedInfo.getEmploy() && CollectionUtils.isNotEmpty(
                processedInfo.getCouponAndTemplateInfos()
        )) {
            LOG.info("Settle User Coupon: {}, {}", info.getUserId(),
                    JSON.toJSONString(settleCoupons));
            // 更新缓存
            redisService.addCouponToCache(
                    info.getUserId(),
                    settleCoupons,
                    CouponStatus.USED.getCode()
            );
            // 更新 db
            kafkaTemplate.send(
                    Constant.TOPIC,
                    JSON.toJSONString(new CouponKafkaMessage(
                            CouponStatus.USED.getCode(),
                            settleCoupons.stream().map(Coupon::getId)
                                    .collect(Collectors.toList())
                    ))
            );
        }

        return processedInfo;
    }

    /**
     * 保留两位小数
     * */
    private double retain2Decimals(double value) {

        // BigDecimal.ROUND_HALF_UP 代表四舍五入
        return new BigDecimal(value)
                .setScale(2, BigDecimal.ROUND_HALF_UP)
                .doubleValue();
    }
}
